use crate::codec::d_codec::{DCodecError, Packet, PacketBytesCodec, Transmission};
use crate::serializer::serialization::Serialized;
use crate::start::dispatcher::PacketDispatcher;
use crate::start::server::{Message, Server, Service, Shared};
use crate::start::{Connection, SendMsg};
use async_trait::async_trait;
use bytes::{BufMut, BytesMut};
use futures::{SinkExt, Stream, StreamExt};
use std::error::Error;
use std::net::SocketAddr;
use std::pin::Pin;
use std::sync::Arc;
use std::task::{Context, Poll};
use std::{io, thread};
use tokio::io::AsyncReadExt;
use tokio::net::{TcpListener, TcpSocket, TcpStream};
use tokio::runtime;
use tokio::runtime::Runtime;
use tokio::sync::mpsc::{UnboundedReceiver, UnboundedSender};
use tokio::sync::{mpsc, Mutex};
use tokio_util::codec::{Decoder, Framed};
use crate::start::processor::Processor;

pub struct TcpServer;

impl TcpServer {
//     /// Process an individual chat client
//     async fn process(
//         state: Arc<Mutex<Shared>>,
//         stream: TcpStream,
//         service: Arc<PacketDispatcher>,
//     ) -> Result<(), Box<dyn Error>> {
//         let framed = Framed::new(stream, PacketBytesCodec::default());
//
//         let mut peer = TcpConnection::new(state.clone(), framed).await?;
//
//         // Process incoming messages until our stream is exhausted by a disconnect.
//         while let Some(result) = peer.next().await {
//             match result {
//                 // A message was received from a peer. Send it to the
//                 // current user.
//                 Ok(Message::Received(msg, addr)) => {
//                     // peer.lines.send(&msg).await?;
//
//                     println!("{:?}", msg);
//                     let connection = Connection {
//                         packet: msg,
//                         send: peer.send.clone(),
//                         addr,
//                     };
//                     let service = service.clone();
//                     tokio::spawn(async move {
//                         //     Dispatcher
//                         service.dispatcher(connection);
//                     });
//                     // peer.framed.send(msg.cover_send());
//                 }
//                 Err(e) => {
//                     tracing::error!(
//                         "an error occurred while processing messages for {:?}; error = {:?}",
//                         addr,
//                         e
//                     );
//                 }
//                 Ok(Message::Broadcast(_, ..)) => {}
//                 Ok(Message::Forward(_, ..)) => {}
//                 Ok(Message::Close()) => break,
//                 Ok(Message::Send(msg, _)) => {
//                     peer.framed.send(msg).await?;
//                     peer.framed.flush().await?;
//                 }
//             }
//         }
//
//         Ok(())
//     }
}

#[async_trait]
impl Server for TcpServer {
    type Socket = TcpListener;

    async fn bind(address: SocketAddr) -> io::Result<Self::Socket> {
        Ok(TcpListener::bind(address).await?)
    }

    async fn poll(socket: Self::Socket, service: Arc<PacketDispatcher>) {
        let state = Arc::new(Mutex::new(Shared::new()));
        loop {
            let (stream, addr) = socket.accept().await.unwrap();
            let state = Arc::clone(&state);
            let service = service.clone();
            tokio::spawn(async move {
                tracing::info!("收到新链接{:?}", thread::current().id());
                tracing::info!("{:?}", &stream);
                let mut tcp_processor =TcpProcessor;
                if let Err(e) = tcp_processor.poll(stream, service).await {
                    tracing::info!("an error occurred; error = {:?}", e);
                }
            });
        }
    }
}

// /// Shorthand for the transmit half of the message channel.
// type Tx = mpsc::UnboundedSender<bool>;
//
// /// Shorthand for the receive half of the message channel.
// type Rx = mpsc::UnboundedReceiver<bool>;

pub struct TcpConnection<U: Decoder> {
    framed: Framed<TcpStream, U>,
    rx: UnboundedReceiver<Message>,
    send: UnboundedSender<Message>,
    addr: SocketAddr,
}

impl<U: Decoder> TcpConnection<U> {
    async fn new(
        // state: Arc<Mutex<Shared>>,
        framed: Framed<TcpStream, U>,
    ) -> io::Result<TcpConnection<U>> {
        // Get the client socket address
        let addr = framed.get_ref().peer_addr()?;

        // Create a channel for this peer
        // let (tx, rx) = mpsc::unbounded_channel();
        let (send, rx) = mpsc::unbounded_channel();

        // Add an entry for this `Peer` in the shared state map.
        // state.lock().await.peers.insert(addr, tx);

        Ok(TcpConnection {
            framed,
            send,
            rx,
            addr,
        })
    }
}

impl<U: Decoder + Decoder<Item = Packet>> Stream for TcpConnection<U> {
    type Item = Result<Message, DCodecError<U>>;

    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
        if let Poll::Ready(Some(v)) = Pin::new(&mut self.rx).poll_recv(cx) {
            //  根据不同的响应消息进行不同的处理
            return Poll::Ready(Some(Ok(v)));
        }
        // 没有响应消息进行下一步接受消息
        // Secondly poll the `Framed` stream.
        let result: Option<_> = futures::ready!(Pin::new(&mut self.framed).poll_next(cx));
        Poll::Ready(match result {
            // We've received a message we should broadcast to others.
            Some(Ok(message)) => Some(Ok(Message::Received(message, self.addr))),

            // An error occurred.
            Some(Err(e)) => Some(Err(DCodecError::Io(e))),

            // The stream has been exhausted.
            None => None,
        })
    }
}


pub struct TcpProcessor;
#[async_trait]
impl Processor for TcpProcessor {
    type Socket = TcpStream;

    async fn poll(&mut self, socket: Self::Socket, service: Arc<PacketDispatcher>)-> Result<(), Box<dyn Error>>  {
        let framed = Framed::new(socket, PacketBytesCodec::default());

        let mut peer = TcpConnection::new( framed).await?;

        // Process incoming messages until our stream is exhausted by a disconnect.
        while let Some(result) = peer.next().await {
            match result {
                // A message was received from a peer. Send it to the
                // current user.
                Ok(Message::Received(msg, addr)) => {
                    // peer.lines.send(&msg).await?;

                    println!("{:?}", msg);
                    let connection = Connection {
                        packet: msg,
                        send: peer.send.clone(),
                        addr,
                    };
                    let service = service.clone();
                    tokio::spawn(async move {
                        //     Dispatcher
                        service.dispatcher(connection);
                    });
                    // peer.framed.send(msg.cover_send());
                }
                Err(e) => {
                    tracing::error!(
                        "an error occurred while processing messages ; error = {:?}",
                        // addr,
                        e
                    );
                }
                Ok(Message::Broadcast(_, ..)) => {}
                Ok(Message::Forward(_, ..)) => {}
                Ok(Message::Close()) => break,
                Ok(Message::Send(msg, _)) => {
                    peer.framed.send(msg).await?;
                    peer.framed.flush().await?;
                }
            }
        }

        Ok(())
    }
}



